package com.xjscrm.server.config;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.uzai.mobile.collect.api.MicaTopics;
import com.uzai.mobile.collect.api.dto.MicaTriggerDto;
import com.uzai.mobile.collect.api.enums.MicaTriggerType;
import com.xjscrm.server.biz.mq.producer.SendMsgProducer;
import com.xjscrm.server.utils.DelayTopicHolder;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

/**
 * @author mingyang.chen create in 2024/3/29 17:09
 */
@Configuration
@Slf4j
public class MqConfig implements InitializingBean {

    @Autowired
    private RocketMQProperties rocketMQProperties;
    @Autowired
    private SendMsgProducer sendMsgProducer;

    @Override
    public void afterPropertiesSet() throws Exception {

        String nameServer = rocketMQProperties.getNameServer();

        for (int i = 0; i < 15; i++) {
            String topic = "delay_xj_topic_" + i;
            String group = "delay_xj_topic_" + i;

            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);

            consumer.setNamesrvAddr(nameServer);
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            consumer.setMessageModel(MessageModel.CLUSTERING);
            //
            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
                try {
                    for (MessageExt msg : msgs) {
                        String s = new String(msg.getBody());
                        MicaTriggerDto micaTriggerDto = JSON.parseObject(s, MicaTriggerDto.class);
                        log.info("--收到延迟队列消息, topic={}, body: {}", topic, JSONObject.toJSONString(micaTriggerDto));
                        MicaTriggerType byValue = MicaTriggerType.getByValue(micaTriggerDto.getMsgType());
                        if (micaTriggerDto.getDelay() > 0) {//还在延迟中，继续丢入延时队列当中
                            log.info("--继续发向延迟队列消息{}，delay={}, micaTransDto={}", byValue.getDesc(), micaTriggerDto.getDelay(), JSONObject.toJSONString(micaTriggerDto));
                            sendMsgProducer.pushMsgDelay(micaTriggerDto, DelayTopicHolder.getNextServer());
                        } else { //立即消费，立即丢入触发指令主题中
                            log.info("--立即处理延迟队列消息{}，delay={}, micaTransDto={}", byValue.getDesc(), micaTriggerDto.getDelay(), JSONObject.toJSONString(micaTriggerDto));
                            sendMsgProducer.pushMsg(micaTriggerDto, MicaTopics.WW_TRIGGER_TOPIC);
                        }
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            });
            consumer.subscribe(topic, MessageSelector.byTag("*"));
            consumer.start();

            log.info("创建延迟 topic: [{}], group: [{}]", topic, group);
            DelayTopicHolder.addDelayTopic(topic);
        }

        log.info("延迟队列创建完成");
    }
}
